package com.cqdev.washer.biz.service.impl;

import com.cqdev.washer.biz.model.TableBO;
import com.cqdev.washer.biz.service.TransferDataService;
import com.cqdev.washer.infra.enums.DBDriverType;
import com.cqdev.washer.infra.service.JdbcTemplateService;
import com.google.common.base.Strings;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author yunfei.li
 *         类描述: 转移数据，删除目标表数据，在插入数据
 *         日期： 2018/7/9
 */
@Service("transferOverwriteDataServiceImpl")
public class TransferOverwriteDataServiceImpl implements TransferDataService {

    private final JdbcTemplateService jdbcTemplateService;

    public TransferOverwriteDataServiceImpl(JdbcTemplateService jdbcTemplateService) {
        this.jdbcTemplateService = jdbcTemplateService;
    }

    @Override
    public int syncTable(TableBO sourceTable, TableBO targetTable) {
        JdbcTemplate sourceJdbcTemplate = jdbcTemplateService.jdbcTemplate(sourceTable.getDbConfigBO());
        JdbcTemplate targetJdbcTemplate = jdbcTemplateService.jdbcTemplate(targetTable.getDbConfigBO());

        List<Map<String, Object>> sourceDataList = findSourceAll(sourceJdbcTemplate, sourceTable.getDbConfigBO().getDbName(), sourceTable.getTableName());
        int batchInsertNum = overwriteTableForSourceData(targetJdbcTemplate, targetTable.getDbConfigBO().getDbName(), targetTable.getTableName(), targetTable.getDbConfigBO().getDbDriverType(), sourceDataList);
        return batchInsertNum;
    }

    private int overwriteTableForSourceData(JdbcTemplate targetJdbcTemplate, final String dbName, final String tableName, DBDriverType dbDriverType, final List<Map<String, Object>> dataList) {
        emptyTable(dbDriverType, targetJdbcTemplate, dbName, tableName);
        int batchNum = writeTable(targetJdbcTemplate, dbName, tableName, dataList);
        return batchNum;

    }

    private int writeTable(JdbcTemplate targetJdbcTemplate, String dbName, String tableName, List<Map<String, Object>> dataList) {
        if (dataList.isEmpty()) {
            return 0;
        }
        Map<String, Object> dataMap = dataList.get(0);
        final Set<String> keySet = dataMap.keySet();

        String sql = buildInsertSql(dbName, tableName, keySet);
        BatchPreparedStatementSetter batchPreparedStatementSetter = new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement preparedStatement, int index) throws SQLException {
                Map<String, Object> valueMap = dataList.get(index);

                String[] keys = keySet.toArray(new String[]{});
                for (int i = 1; i <= keys.length; i++) {
                    preparedStatement.setObject(i, valueMap.get(keys[i - 1]));
                }
            }

            @Override
            public int getBatchSize() {
                return dataList.size();
            }
        };

        int[] batchUpdate = targetJdbcTemplate.batchUpdate(sql, batchPreparedStatementSetter);
        return batchUpdate.length;
    }

    private void emptyTable(DBDriverType dbDriverType, JdbcTemplate targetJdbcTemplate, String dbName, String tableName) {
        String table = dbName + "." + tableName;
        String empty_sql = null;
        switch (dbDriverType) {
            case MYSQL:
                empty_sql = "DELETE FROM " + table;
                break;
            case HIVE:
                empty_sql = "INSERT OVERWRITE TABLE " + table + " SELECT * FROM " + table + " WHERE 1 = 0";
                break;
        }
        targetJdbcTemplate.execute(empty_sql);
    }

    private String buildInsertSql(String dbName, String tableName, Set<String> keySet) {
        if (keySet.isEmpty()) {
            throw new RuntimeException("build sql exception");

        }

        StringBuilder sqlBuilder = new StringBuilder();
        sqlBuilder.append("INSERT INTO ");
        sqlBuilder.append(dbName);
        sqlBuilder.append(".");
        sqlBuilder.append(tableName);
        sqlBuilder.append("(");

        String[] keys = keySet.toArray(new String[]{});
        for (int i = 0; i < keys.length; i++) {
            String key = keys[i];
            if (key.contains(".")) {
                key = key.split("\\.")[1];
            }
            sqlBuilder.append(key);
            if ((keys.length - 1) != i) {
                sqlBuilder.append(",");
            }
        }
        sqlBuilder.append(")");
        sqlBuilder.append("VALUES (");

        for (int i = 0; i < keys.length; i++) {
            sqlBuilder.append("?");
            if ((keys.length - 1) != i) {
                sqlBuilder.append(",");
            }
        }
        sqlBuilder.append(")");
        return sqlBuilder.toString();
    }

    private List<Map<String, Object>> findSourceAll(JdbcTemplate sourceJdbcTemplate, String dbName, String tableName) {
        if (Strings.isNullOrEmpty(dbName) || Strings.isNullOrEmpty(tableName)) {
            throw new IllegalArgumentException("param illegal.");
        }
        List<Map<String, Object>> dataList = sourceJdbcTemplate.queryForList("SELECT * FROM " + dbName + "." + tableName);
        return dataList;
    }
}
